from pyspark import SparkConf ,SparkContext

conf = SparkConf().setMaster("local").setAppName("RDDKeyAndValue")
sc = SparkContext(conf= conf)

'''
# 第一种创建方式:从文件中加载
# 可以采用多种方式创建键值对 RDD ,其中一种主要方式是使用 map() 函数来实现
lines = sc.textFile("file:///home/hadoop/MyTmp/sparkwordcount.txt")
pairRDD = lines.flatMap(lambda line:line.split(" ")).map(lambda word:(word,1))
pairRDD.foreach(print)

'''

'''
# 第二种创建方式:通过并行集合(列表)创建 RDD
list = ["hadoop","spark","Hive","spark"]
rdd = sc.parallelize(list)
pairRDD = rdd.map(lambda word:(word,1))
pairRDD.foreach(print)
'''

'''
# reduceByKey(func) Demo
pairRDD = sc.parallelize([("Hadoop",1),("Spark",1),("Hive",1),("Spark",1)])
pairRDD.reduceByKey(lambda a,b:a+b)
pairRDD.foreach(print)
'''

'''
# groupByKey() 的功能是,对具有相同键的值进行分组
list = [("spark",1),("spark",2),("hadoop",3),("hadoop",4)]
pairRDD = sc.parallelize(list)
print("result1:",pairRDD.groupByKey())
print("*"*30)
pairRDD.groupByKey().foreach(print)
'''

'''
# keys
# keys 只会把 Pair RDD 中的 key 返回形成一个新的 RDD
list = [("Hadoop",1),("Spark",1),("Hive",1),("Spark",1)]
pairRDD = sc.parallelize(list)
pairRDD.keys().foreach(print)
'''

'''
# • values
# values 只会把 Pair RDD 中的 value 返回形成一个新的 RDD 。
list = [("Hadoop",1),("Spark",1),("Hive",1),("Spark",1)]
pairRDD = sc.parallelize(list)
pairRDD.values().foreach(print)
'''

'''
# sortByKey()
# sortByKey() 的功能是返回一个根据键排序的 RDD
list = [("Hadoop",1),("Spark",1),("Hive",1),("Spark",1)]
pairRDD = sc.parallelize(list)
pairRDD.foreach(print)
print("*"*20)
pairRDD.sortByKey().foreach(print)

'''


'''
# mapValues(func)
# 对键值对 RDD 中的每个 value 都应用一个函数,但是, key 不会发生变化
list = [("Hadoop",1),("Spark",1),("Hive",1),("Spark",1)]
pairRDD = sc.parallelize(list)
pairRDD1 = pairRDD.mapValues(lambda x:x+1)
pairRDD1.foreach(print)
'''

'''
# join
# join 就表示内连接。对于内连接,对于给定的两个输入数据集 (K,V1) 和 (K,V2) ,只有在两个数据集中都存在的 key 才会被输出,最终得到一个 (K,(V1,V2)) 类型的数据集
pairRDD1 = sc.parallelize([("spark",1),("spark",2),("hadoop",3),("hadoop",5)])
pairRDD2 = sc.parallelize([("spark","fast")])
pairRDD3 = pairRDD1.join(pairRDD2)
pairRDD3.foreach(print)
'''

'''
pairRDD1 = sc.parallelize([("spark",1),("spark",2),("hadoop",3),("hadoop",5)])
pairRDD2 = sc.parallelize([("spark","fast")])
pairRDD3 = pairRDD1.subtractByKey(pairRDD2)
pairRDD3.foreach(print)
'''

# 题目:给定一组键值对 ("spark",2),("hadoop",6),("hadoop",4),("spark",6) ,键值对的 key 表示图书名称, value 表示某天图书销量,请计算每个键对应的平均值,也就是计算每种图书的每天平均销量。
rdd = sc.parallelize([("spark",2),("hadoop",6),("hadoop",4),("spark",6)])
rddResult = rdd.mapValues(lambda x:(x,1)).\
    reduceByKey(lambda x,y:(x[0]+y[0],x[1]+y[1])).\
    mapValues(lambda x:x[0]/x[1])
print(rddResult.collect())
